package com.zengqingfa.rocketmq.transaction;

import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.spring.annotation.RocketMQTransactionListener;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionListener;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionState;
import org.apache.rocketmq.spring.support.RocketMQHeaders;
import org.springframework.messaging.Message;

import java.util.HashMap;
import java.util.Map;

/**
 *
 * @fileName: TransactionListenerImpl
 * @author: zengqf3
 * @date: 2021-4-3 11:49
 * @description:
 *
 * 发送事务消息:
 * 在客户端，首先用户需要实现RocketMQLocalTransactionListener接口，并在接口类上注解声明 @RocketMQTransactionListener，
 * 实现确认和回查方法；然后再使用资源模板 RocketMQTemplate， 调用方法sendMessageInTransaction()来进行消息的发布。
 * 注意：从 RocketMQ-Spring 2.1.0版本之后，注解@RocketMQTransactionListener不能设置 txProducerGroup、ak、sk，
 * 这些值均与对应的RocketMQTemplate保持一致。
 *
 * RocketMQTransactionListener :2.1.0版本 以上实现了使用同一个TransactionMQProducer来发送所有类型的事务消息
 * 当我使用RocketMQTemplate并发的执行事务时，非常容易重现类似"illegal state"的异常
 * 这个问题的原因是一个TransactionProducer在执行事务时不能被共享 所以，我们不得不保证一个TransactionProducer
 * 实例必须不能被并发使用
 */
//@RocketMQTransactionListener
@Slf4j
public class TransactionListenerImpl implements RocketMQLocalTransactionListener {

    private static Map<String, RocketMQLocalTransactionState> STATE_MAP = new HashMap<>();

    /**
     *  执行业务逻辑
     *
     * @param message
     * @param o
     * @return
     */
    @Override
    public RocketMQLocalTransactionState executeLocalTransaction(Message message, Object o) {
        String transId = (String) message.getHeaders().get(RocketMQHeaders.TRANSACTION_ID);
        try {
            log.info("开始执行本地事务->>>>>>>>>");
            Thread.sleep(500);
            log.info("本地事务执行结束->>>>>>>>>");
        } catch (Exception e) {
            log.info("本地事务执行失败,e:{}", e.getMessage());
            STATE_MAP.put(transId, RocketMQLocalTransactionState.ROLLBACK);
            return RocketMQLocalTransactionState.ROLLBACK;
        }
        // 设置事务状态
        STATE_MAP.put(transId, RocketMQLocalTransactionState.COMMIT);
        // 返回事务状态给生产者
        return RocketMQLocalTransactionState.COMMIT;
    }

    /**
     * 回查
     *
     * @param message
     * @return
     */
    @Override
    public RocketMQLocalTransactionState checkLocalTransaction(Message message) {
        log.info("开始消息回查->>>>>>>>>");
        String transId = (String) message.getHeaders().get(RocketMQHeaders.TRANSACTION_ID);
        log.info("回查消息 -> transId= {},state = {} ", transId, STATE_MAP.get(transId));
        return STATE_MAP.get(transId);
    }
}
